package com.cit.draw.redis;


import java.util.Set;

import redis.clients.jedis.BinaryJedis;

import com.google.common.base.Preconditions;
import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.google.inject.name.Named;

public class PersistedCacheRequestHandlerImpl implements
		PersistedCacheRequestHandler {

	/**
	 * Seconds in a minute, used to convert expiration time in minutes.
	 */
	private static final int MINUTE = 60;

	/**
	 * Value to avoid cached value expiration.
	 */
	private static final int NO_EXPIRES = Integer.MAX_VALUE;

	@Inject
	protected BinaryJedisPool jedisPool;

	@Override
	public <T extends PersistedCacheObject> void putPersistedCache(T entity,
			String key) {
		putPersistedCache(entity, key, NO_EXPIRES);
	}

	@Override
	public <T extends PersistedCacheObject> void putPersistedCache(T entity,
			String key, int expirationMinutes) {
		Preconditions.checkNotNull(jedisPool,
				"Critical: Master Jedis can not be null!");
		Preconditions.checkNotNull(entity);
		Preconditions.checkNotNull(key);
		Preconditions.checkArgument(expirationMinutes >= 0,
				"Expiration time must be positive or 0");
		if (expirationMinutes == 0) {
			// this is to make the record expired
			// and also no more put in the cache
			deletePersistedCache(key);
			return;
		}

		EntitySerializer serializer = new EntitySerializer();
		byte[] value = serializer.serialize(entity);

		BinaryJedis jedis = null;
		try {

			jedis = jedisPool.getResource();

			if (jedis != null && value != null) {
				if (expirationMinutes == NO_EXPIRES) {
					jedis.set(key.getBytes(), value);
				} else {
					jedis.setex(key.getBytes(), expirationMinutes * MINUTE,
							value);
				}
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T extends PersistedCacheObject> T getPersistedCache(String key) {
		Preconditions.checkNotNull(jedisPool,
				"Critical: Master Jedis can not be null!");
		Preconditions.checkNotNull(key);

		EntitySerializer serializer = new EntitySerializer();

		BinaryJedis jedis = null;
		byte[] value = null;

		try {
			jedis = jedisPool.getResource();
			if (!jedis.exists(key.getBytes())) {
				return null;
			}
			value = jedis.get(key.getBytes());
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}

		}

		return (T) serializer.deserialize(value);

	}

	@Override
	public <T extends PersistedCacheObject> void deletePersistedCache(String key) {
		Preconditions.checkNotNull(jedisPool,
				"Critical: Master Jedis can not be null!");
		Preconditions.checkNotNull(key);

		BinaryJedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			if (jedis.exists(key.getBytes())) {
				jedis.del(key.getBytes());
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}

		}

	}

	@Override
	public Long increase(String key) {
		Preconditions.checkNotNull(key);

		byte[] keyByte = key.getBytes();
		BinaryJedis jedis = null;
		Long result = -1L;
		try {
			jedis = jedisPool.getResource();
			synchronized (key) {
				result = jedis.incr(keyByte);
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}

		}

		return result;
	}

	@Override
	public Long decrease(String key) {
		Preconditions.checkNotNull(key);

		byte[] keyByte = key.getBytes();
		BinaryJedis jedis = null;
		Long result = -1L;
		try {
			jedis = jedisPool.getResource();
			synchronized (key) {
				result = jedis.decr(keyByte);
			}

		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}

		}

		return result;
	}

	@Override
	public Long get(String key) {
		Preconditions.checkNotNull(key);

		byte[] keyByte = key.getBytes();
		BinaryJedis jedis = null;
		Long result = -1L;
		try {
			jedis = jedisPool.getResource();
			// result = Long.parseLong(new String(jedis.get(keyByte)));
			synchronized (key) {
				result = jedis.incrBy(keyByte, 0);
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}

		}

		return result;
	}

	@Override
	public <T> void putPrimitivePersistedCache(T entity, String key) {
		putPrimitivePersistedCache(entity, key, NO_EXPIRES);

	}

	@Override
	public <T> void putPrimitivePersistedCache(T entity, String key,
			int expirationMinute) {
		Preconditions.checkNotNull(jedisPool,
				"Critical: Master Jedis can not be null!");
		Preconditions.checkNotNull(entity);
		Preconditions.checkNotNull(key);
		Preconditions.checkArgument(expirationMinute >= 0,
				"Expiration time must be positive or 0");
		if (expirationMinute == 0) {
			// this is to make the record expired
			// and also no more put in the cache
			deletePersistedCache(key);
		}

		// Serialize
		String valueString = entity.toString();
		byte[] value = valueString.getBytes();
		// initial the binary Jedis
		BinaryJedis jedis = null;
		try {

			jedis = jedisPool.getResource();

			if (jedis != null && value != null) {
				if (expirationMinute == NO_EXPIRES) {
					jedis.set(key.getBytes(), value);
				} else {
					jedis.setex(key.getBytes(), expirationMinute * MINUTE,
							value);
				}
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}

	}

	@Override
	public String getPrimitivePersistedCache(String key) {
		Preconditions.checkNotNull(jedisPool,
				"Critical: Master Jedis can not be null!");
		Preconditions.checkNotNull(key);

		byte[] keyByte = key.getBytes();
		BinaryJedis jedis = null;
		byte[] result = null;
		try {
			jedis = jedisPool.getResource();
			synchronized (key) {
				result = jedis.get(keyByte);
			}
			if (result == null) {
				return null;
			}

			return new String(result);

		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}
	}

	@Override
	public void flush() {
		BinaryJedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			jedis.flushAll();

		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}
	}

	@Override
	public void set(String key, Long value) {
		Preconditions.checkNotNull(key);
		Preconditions.checkNotNull(value);

		byte[] keyBytes = key.getBytes();
		BinaryJedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			synchronized (key) {
				jedis.set(keyBytes, value.toString().getBytes());
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}

	}

	@Override
	public Set<String> getSetCache(String username) {
		Preconditions.checkNotNull(username);

		BinaryJedis jedis = null;
		Set<String> result = Sets.newHashSet();

		try {
			jedis = jedisPool.getResource();
			synchronized (username) {
				Set<byte[]> smembers = jedis.smembers(username.getBytes());
				for (byte[] keyBytes : smembers) {
					result.add(new String(keyBytes));
				}
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}

		if(result.isEmpty()){
			return null;
		}
		
		return result;

	}

	@Override
	public void putSetCache(String username, Set<String> list) {
		Preconditions.checkNotNull(username);
		Preconditions.checkNotNull(list);
		Preconditions.checkArgument(!list.isEmpty());

		BinaryJedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			synchronized (username) {
				if (jedis.scard(username.getBytes()) == 0) {
					// new entry
					for (String key : list) {
						jedis.sadd(username.getBytes(), key.getBytes());
					}
				} else {
					// existed set
					for (String key : list) {
						if (jedis
								.sismember(username.getBytes(), key.getBytes())) {
							continue;
						}

						jedis.sadd(username.getBytes(), key.getBytes());
					}
				}
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}

	}

	@Override
	public void addCacheKeyIntoKeySet(String username, String key) {
		Preconditions.checkNotNull(username);
		Preconditions.checkNotNull(key);

		BinaryJedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			synchronized (username) {
				// existed set
				if (jedis.sismember(username.getBytes(), key.getBytes())) {
					return;
				}
				jedis.sadd(username.getBytes(), key.getBytes());
			}
		} finally {
			if (jedis != null) {
				jedisPool.returnResource(jedis);
			}
		}
	}

}
